package com.wlyuan.open.scanner.application.listener.message.consumer;

import com.alibaba.fastjson.JSON;
import com.wlyuan.open.scanner.application.listener.table.TableListener;
import com.wlyuan.open.scanner.domain.message.MessageListenerParse;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import lombok.var;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.stereotype.Service;

import java.util.Objects;

/**
 * @author lijing
 * @description
 */
@Slf4j
@Service
@Configuration
@RequiredArgsConstructor
public class TableMessageConsumer {

    private final TableListener tableListener;

    @Bean
    public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
        return ((message, exception, consumer) -> {
            logger.error("监听增量消息异常:{},原因:{},{}", message.getPayload().toString(), exception.getMessage(), exception.getCause().getLocalizedMessage());
            return null;
        });
    }

    /**
     * 监听消息 消息体 {"before": {},"after": {},"source": {},"op": "","ts_ms": null,"transaction": null}
     * topicPattern = "sync.dbo.*"
     *
     * @param message
     */
    //@KafkaListener(topics = "#{topicHandler.getTopics()}", errorHandler = "consumerAwareErrorHandler")
    public void consumer(String message) {
        logger.info("************************************************************");
        logger.info("接收到消息:{}", message);
        var parse = MessageListenerParse.parse(message);
        if (Objects.isNull(parse)) {
            logger.info("消息解析失败");
            return;
        }
        logger.info("解析消息后:{}", JSON.toJSON(parse));
        tableListener.schema(parse.getId(), parse.getTable(), parse.getData(), parse.getBeforeData());
    }


}
